using System;
using System.Collections.Generic;
using System.Text;
using System.Linq;
using SimpleServiceBus.Bus;
using SimpleServiceBus.Bus.MessageManagement;
using SimpleServiceBus.Bus.Pipeline;
using Common.Logging;
using System.Globalization;

namespace SimpleServiceBus.Endpoints.ActiveMQ.Transport {

    public class ActiveMQMessageBus: MessageBus, IExtendedMessageBus {

        //-----< Fields >-----
        //----------------------------------------------------------------------------------        
        private static readonly ILog fLog = LogManager.GetLogger(typeof(MessageBus));
        private IMessageTypeManagementService fMessageTypeManagementService;


        //-----< Constructors >-----
        //----------------------------------------------------------------------------------        
        public ActiveMQMessageBus() {
        }

        //-----< Public and protected properties >-----
        //----------------------------------------------------------------------------------        

        protected IMessageTypeManagementService MessageTypeManagementService {
            get {
                if(fMessageTypeManagementService == null) {
                    fMessageTypeManagementService = ServiceManager.GetService<IMessageTypeManagementService>();
                    if(fMessageTypeManagementService == null)
                        throw new Exception(("no IMessageTypeManagementService was registered"));
                }
                return fMessageTypeManagementService;
            }
        }

        //-----< Public and protected methods >-----
        //----------------------------------------------------------------------------------        

        public object Ask(object request, TimeSpan timeout)
        {
            //Normal ASK without forced destination ( will cause lookup as usual)
            return Ask(request, timeout, null);
        }

        public Object Ask(Object request, TimeSpan timeout, string Destination)
        {
            CheckRunning("Ask request");
            var _transport = (ActiveMQTransport)this.Transport;
            var _responseEnvelope = InternalSend(request, (envelope, destination) => _transport.Ask(envelope, destination, timeout),Destination);
            if (_responseEnvelope != null)
            {
                try
                {
                    //Pipeline patch  as seen:
                    // http://simpleservicebus.codeplex.com/discussions/233332
                    //for handeling more then one ReceivePipeline per thread
                    var _currentReceivePipeline = MessagePipeline.CurrentReceivePipeline;
                    if (_currentReceivePipeline != null)
                        MessagePipeline.ClearCurrentPipeline(PipelineDirection.Receive);
                    try
                    {
                        return ProcessRequestResponse(_responseEnvelope);
                    }
                    finally
                    {
                        if (_currentReceivePipeline != null)
                            MessagePipeline.SetCurrentPipeline(_currentReceivePipeline, PipelineDirection.Receive);
                    }
                }
                catch (Exception ex)
                {
                    fLog.Error("Unhandled error processing message", ex);
                    throw;
                }
            }
            return null;
        }

        /// <summary>
        /// Processing a requestResponse for allowing more then one pipeline request 
        /// </summary>
        /// <param name="responseEnvelope"></param>
        /// <returns></returns>
        private Object ProcessRequestResponse(MessageEnvelope responseEnvelope)
        {
            var _factory = new ActiveMQMessagePipelineFactory();
            _factory.Endpoint = this.Endpoint;
            _factory.IgnoreMessageDispatch = true;
            var _pipeline = _factory.CreateReceivePipelineFor(responseEnvelope);

            OnBeginReceiveMessage(responseEnvelope, _pipeline);

            _pipeline.PipelineEvents.PipelineError += Events_PipelineError;
            _pipeline.ProcessMessage();

            OnMessageReceived(responseEnvelope);

            return responseEnvelope.Body;
        }



        protected override MessageEnvelope GetMessageEnvelopeFor(object message, params string[] destinations) {
            var result = new ActiveMQMessageEnvelope {
                MessageID = GetNextMessageID(),
                Body = message,
                ReturnAddress = GetReturnAddressFor(destinations)
            };

            result.Recoverable = RecoverableAttribute.IsRecoverable(message);

            return result;
        }

        protected MessageEnvelope InternalSend(object message, Func<MessageEnvelope, String, MessageEnvelope> sendFunc)
        {
            //we use the override which can add a destination. If the destination is null, we lookup
            return InternalSend(message, sendFunc, null);
        }

        protected MessageEnvelope InternalSend(object message, Func<MessageEnvelope, String, MessageEnvelope> sendFunc,string Destination)
        {
            var _transport = (ActiveMQTransport)this.Transport;
            try
            {
                
                //We override the lookup for the message to force a destination injection so we can pick the destination where to send it..;
                string destination = Destination;
                if (string.IsNullOrEmpty(destination))
                {
                    //try to look it up
                    destination = MessageTypeManagementService.GetDestinationForMessage(message);
                }
                if (string.IsNullOrEmpty(destination))
                    throw new Exception(("A destination could not be located for this type of message:" + message.GetType().Name));
                MessageEnvelope toSend;

                if (message is MessageEnvelope)
                    toSend = (MessageEnvelope)message;
                else
                    toSend = GetMessageEnvelopeFor(message, destination);

                var factory = ServiceManager.GetService<IMessagePipelineFactory>();
                if (factory != null)
                {
                    IMessagePipeline pipeline = factory.CreatePipelineFor(toSend, PipelineDirection.Send);
                    if (pipeline != null)
                    {
                        OnBeginSendMessage(toSend, pipeline);

                        pipeline.ProcessMessage();
                        if (pipeline.CurrentContext.AbortMessageProcessing)
                        {
                            return null;
                        }

                    }
                }

                var _result = sendFunc(toSend, destination);

                OnMessageSent(toSend);

                return _result;
            }
            catch (Exception ex)
            {
                fLog.Error("Error sending message", ex);
                throw;
            }
        }


      

        public override SimpleServiceBus.Bus.MessageManagement.Callbacks.ICallback Send(object message, params string[] destinations) {
            CheckRunning("Send message");
            return base.Send(message, destinations);
        }

        public void Send(IEnumerable<object> messages) {
            CheckRunning("Send messages");
            var _transport = (ActiveMQTransport)this.Transport;
            using(var _scope = _transport.CreateScope()) {
                try {
                    foreach(var _message in messages) {
                        InternalSend(_message,
                            (envelope, destination) => {
                                _transport.Send(_scope, envelope, destination);
                                return null;
                            });
                    }
                    _scope.Commit();
                }
                catch {
                    _scope.Rollback();
                    throw;
                }
            }
        }

        protected override void ShutDownService() {
            base.ShutDownService();
            fMessageTypeManagementService = null;
        }

        //-----< Events >-----
        //----------------------------------------------------------------------------------        


        //-----< Internal members >-----
        //----------------------------------------------------------------------------------        


        //-----< Private members >-----
        //----------------------------------------------------------------------------------        

        private void CheckRunning(String operation) {
            if(!IsRunning)
                throw new InvalidOperationException(String.Format(CultureInfo.InvariantCulture, "Cannot perform '{0}' because the  endpoint is not running", operation));
        }

        //-----< All nested types >-----
        //----------------------------------------------------------------------------------        


        //-----< All explicit interface implementations >-----
        //----------------------------------------------------------------------------------        



    }
}

